package org.caw.hbase.explorer.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.caw.domain.Row;
import org.caw.domain.RowList;
import org.caw.exception.CawException;
import org.caw.hbase.HBaseTablePool;
import org.caw.hbase.explorer.ScanManager;
import org.caw.hbase.explorer.filter.HBaseFilterParser;
import org.caw.util.ConvertUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Component for scanning HBASE tables
 * 
 * @author dino.keco
 * 
 */
public class HBaseScanManagerImpl implements ScanManager {

	private static final Integer DEFAULT_LIMIT = 5;

	private static final Logger LOGGER = LoggerFactory.getLogger(HBaseScanManagerImpl.class);

	private HBaseTablePool pool;

	private HBaseFilterParser parser;

	public HBaseTablePool getPool() {
		return pool;
	}

	public void setPool(HBaseTablePool pool) {
		this.pool = pool;
	}

	public HBaseFilterParser getParser() {
		return parser;
	}

	public void setParser(HBaseFilterParser parser) {
		this.parser = parser;
	}

	@Override
	public RowList scan(String tableName, String startKey, String endKey, List<String> columns, Integer maxversion, Integer limit) throws CawException {
		Scan scan = ConvertUtil.buildScan(startKey, endKey, columns, maxversion, null);
		scan.setFilter(new PageFilter(limit != null ? limit : DEFAULT_LIMIT));
		return executeScan(tableName, scan, columns);
	}

	@Override
	public RowList scan(String tableName, String startKey, String endKey, List<String> columns, Integer maxversion, Integer limit, String filters) throws CawException {
		Scan scan = ConvertUtil.buildScan(startKey, endKey, columns, maxversion, null);
		Filter filter = parser.parse(filters);
		Filter pageFilter = new PageFilter(limit != null ? limit : DEFAULT_LIMIT);
		FilterList completeFilterList = new FilterList(Operator.MUST_PASS_ALL);
		completeFilterList.addFilter(filter);
		completeFilterList.addFilter(pageFilter);
		scan.setFilter(completeFilterList);
		return executeScan(tableName, scan, columns);
	}

	@Override
	public RowList scan(String tableName, String startKey, String endKey, List<String> columns, Long timestamp, Integer limit) throws CawException {
		Scan scan = ConvertUtil.buildScan(startKey, endKey, columns, 1, timestamp);
		scan.setFilter(new PageFilter(limit != null ? limit : DEFAULT_LIMIT));
		return executeScan(tableName, scan, columns);
	}

	@Override
	public RowList scan(String tableName, String startKey, String endKey, List<String> columns, Long timestamp, Integer limit, String filters) throws CawException {
		Scan scan = ConvertUtil.buildScan(startKey, endKey, columns, 1, timestamp.longValue());
		Filter filter = parser.parse(filters);
		Filter pageFilter = new PageFilter(limit != null ? limit : DEFAULT_LIMIT);
		FilterList completeFilterList = new FilterList(Operator.MUST_PASS_ALL);
		completeFilterList.addFilter(filter);
		completeFilterList.addFilter(pageFilter);
		scan.setFilter(completeFilterList);
		return executeScan(tableName, scan, columns);
	}

	private RowList executeScan(String tableName, Scan s, List<String> columns) throws CawException {
		HTableInterface table = pool.getTable(tableName);
		try {

			ResultScanner rs = table.getScanner(s);
			RowList rowList = new RowList();

			if (columns != null) {
				rowList.setFamilyNames(ConvertUtil.filterQualifiers(columns));
			} else {
				rowList.setFamilyNames(pool.getTableFamilies(tableName));
			}

			rowList.setRows(new ArrayList<Row>());
			for (Result r : rs) {
				Row row = ConvertUtil.convert(r);
				rowList.getRows().add(row);
			}

			return rowList;

		} catch (IOException e) {
			LOGGER.error("Failed to execute scan. Reason: {}", e.getMessage(), e);
			throw new CawException("Failed to execute scan. Reason: " + e.getMessage(), e);
		} finally {
			pool.putTable(table);
		}
	}

}
